<?php

namespace App\Console\Commands;

use App\Events\Event;
use App\Events\In\BillingFinished;
use App\Events\In\CashBillingFinished;
use App\Events\In\NormalBuyingFinished;
use App\Events\In\StudentAppointmentStatusChanged;
use App\Events\In\TeacherAppointmentCreated;
use App\Events\In\TeacherAppointmentStatusChanged;
use App\Events\In\GiftBillingFinished;
use App\Services\AliyunLogService;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Log;

class ConsumeAliyunKafkaMQCommand extends Command
{
    protected $signature = 'aliyunmq:consume-kafka';
    protected $interestedTags = [
        NormalBuyingFinished::TAG => NormalBuyingFinished::class,
        StudentAppointmentStatusChanged::TAG => StudentAppointmentStatusChanged::class,
        BillingFinished::TAG => BillingFinished::class,
        CashBillingFinished::TAG => CashBillingFinished::class,
        GiftBillingFinished::TAG => GiftBillingFinished::class,
    ];

    public function handle()
    {
        $conf = new \RdKafka\Conf();

        $conf->set('sasl.mechanisms', 'PLAIN');
        $conf->set('api.version.request', 'true');
        $conf->set('sasl.username', config('kafka.sasl_plain_username'));
        $conf->set('sasl.password', config('kafka.sasl_plain_password'));
        $conf->set('security.protocol', 'PLAINTEXT');
        $conf->set('ssl.ca.location', __DIR__ . '/../../../config/ca-cert');

        $conf->set('group.id', config('kafka.consumer_id'));

        $conf->set('metadata.broker.list', config('kafka.bootstrap_servers'));

        $topicConf = new \RdKafka\TopicConf();

        $topicConf->set('auto.offset.reset', 'smallest');

        $conf->setDefaultTopicConf($topicConf);

        $consumer = new \RdKafka\KafkaConsumer($conf);

        $consumer->subscribe([config('kafka.topic_name')]);
        /**
         * @var $logService AliyunLogService
         */
        $logService = app('AliyunLog');
        while (true) {
            $message = $consumer->consume(120 * 1000);
            try {
                $this->handleMessagesOnce($message);
            } catch (\Exception $e) {
                Log::error($e->getMessage());
            }
            //TODO: 每次push 是不是太heavy
            $logService->push();
        }
    }

    public function handleMessagesOnce($message)
    {
        switch ($message->err) {
            case \RD_KAFKA_RESP_ERR_NO_ERROR:
                if (empty($message) || empty($message->payload)) {
                    return;
                }
                $messageBody = json_decode($message->payload, true);
                if (array_key_exists('tag', $messageBody) && array_key_exists('body', $messageBody)) {
                    if ($this->interested($messageBody['tag'])) {
                        event($this->parseEvent($messageBody, $message->key));
                    }
                }
                break;
            case \RD_KAFKA_RESP_ERR__PARTITION_EOF:
                echo "No more messages; will wait for more\n";
                break;
            case \RD_KAFKA_RESP_ERR__TIMED_OUT:
                echo "Timed out\n";
                break;
            default:
//                throw new \Exception($message->errstr(), $message->err);
                Log::error('AliyunMQService->Market.Event error [message]' . $message->errstr() . ' [code] ' . $message->err);
                break;
        }
    }

    /**
     * @param $message
     * @return bool
     */
    protected function interested($message)
    {
        return array_key_exists($message, $this->interestedTags);
    }

    protected function parseEvent($message, $key)
    {
        $eventClassName = $this->interestedTags[$message['tag']];
        /**
         * @var $event Event
         */
        $event = new $eventClassName();
        Log::info('[AliyunMQService->Market.Event:interestedMessage] [MessageKey] ' . $key .
            ' [tag] ' . $message['tag'] . ' [body] ' . $message['body']);
        $messageBody = json_decode($message['body']);
        foreach ($messageBody as $key => $value) {
            if (property_exists($event, $key))
                $event->{$key} = $value;
        }

        return $event;
    }
}
